package com.bkjk.utils;

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.pentaho.di.core.Condition;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.gui.Point;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.datagrid.DataGridMeta;
import org.pentaho.di.trans.steps.joinrows.JoinRowsMeta;
import org.pentaho.di.trans.steps.mongodbinput.MongoDbInput;
import org.pentaho.di.trans.steps.mongodbinput.MongoDbInputData;
import org.pentaho.di.trans.steps.mongodbinput.MongoDbInputMeta;
import org.pentaho.di.trans.steps.mongodboutput.MongoDbOutputMeta;
import com.bkjk.constant.DataBaseMetaConstant;

public class KettleUtils {
	/**
	 * 连接mongo默认使用的加密方式
	 */
	private final static String authenticationMechanism = "SCRAM-SHA-1";
	
	

	/**
	 * 
	 * /** This is the base step that forms that basis for all steps. You can
	 * derive from this class to implement your own steps.
	 *
	 * @param stepMeta
	 *            The StepMeta object to run.
	 * @param stepDataInterface
	 *            the data object to store temporary data, database connections,
	 *            caches, result sets, hashtables etc.
	 * @param copyNr
	 *            The copynumber for this step.
	 * @param transMeta
	 *            The TransInfo of which the step stepMeta is part of.
	 * @param trans
	 *            The (running) transformation to obtain information shared
	 *            among the steps.
	 */

	public static MongoDbInput getMongoInput(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
			TransMeta transMeta, Trans trans) {
		/* DatabaseMeta databaseMeta = new DatabaseMeta(); */

		/* DatabaseMeta.TYPE_DATABASE */
		MongoDbInput mongoDbInput = new MongoDbInput(stepMeta, stepDataInterface, copyNr, transMeta, trans);
		return mongoDbInput;
	}

	/**
	 * mongo输入组件，需要放入StepMeta组件中
	 * @return
	 */
	public static MongoDbInputMeta getMongoDbInputMeta() {
		MongoDbInputMeta mongoDbInputMeta = new MongoDbInputMeta();
		mongoDbInputMeta.setAuthenticationMechanism(authenticationMechanism);
		return mongoDbInputMeta;
	}
	/**
	 * mongo输出组件
	 * @return
	 */
	public static MongoDbOutputMeta getMongDbOutputMeta() {
		MongoDbOutputMeta mongoDbOutputMeta = new MongoDbOutputMeta();
		mongoDbOutputMeta.setAuthenticationMechanism(authenticationMechanism);
		return mongoDbOutputMeta;
	}

	public static Trans getTrans() {
		Trans trans = new Trans();
		return trans;
	}

	/**
	 * 创建一个转化元数据，包括数据库连接等
	 */
	public static TransMeta getTransMeta(String name) {

		TransMeta transMeta = new TransMeta();
		transMeta.setName(name);
		return transMeta;
	}

	/**
	 * 创建一个转化元数据，包括数据库连接等
	 */
	public static TransMeta getTransMeta() {

		TransMeta transMeta = new TransMeta();
		return transMeta;
	}

	/**
	 * 创建一个步骤,默认该步骤为显示
	 */
	public static StepMeta getStepMeta(Point point) {
		// 给步骤添加在spoon工具中的显示位置
		StepMeta stepMeta = new StepMeta();
		stepMeta.setLocation(point);
		stepMeta.setDraw(true);
		return stepMeta;
	}

	/**
	 * 创建一个步骤,默认该步骤为显示 默认指定一个位置
	 * ,一定要设置name，不然会报错
	 */
	public static StepMeta getStepMeta() {
		// 给步骤添加在spoon工具中的显示位置
		StepMeta stepMeta = new StepMeta();
		stepMeta.setDraw(true);
		stepMeta.setName("");
		return stepMeta;
	}

	public static Point getPoint(int x, int y) {
		Point point = new Point(x, y);
		return point;
	}

	public static DatabaseMeta getDataBase() {
		DatabaseMeta dbm = new DatabaseMeta();
		dbm.setDatabaseType(DataBaseMetaConstant.DATATYPE_MYSQL);
		dbm.setHostname(DataBaseMetaConstant.HOST);
		// 0代表native
		dbm.setAccessType(0);

		dbm.setDBName(DataBaseMetaConstant.DBNAME);
		dbm.setDBPort(DataBaseMetaConstant.DBPORT);
		dbm.setUsername(DataBaseMetaConstant.USERNAME);
		dbm.setPassword(DataBaseMetaConstant.PASSWORD);
		/* dbm.setName("mysql"); */
		// dbm.setAttributes(attributes);
		return dbm;
	}

	/**
	 * 得到一个常量的数据组件,并设置一些初始值，不设置会报空指针
	 * @return
	 */
	public static DataGridMeta getDataGridMeta(){
		DataGridMeta dataGridMeta = new DataGridMeta();
		dataGridMeta.setFieldFormat(new String[]{""});
		dataGridMeta.setDecimal(new String[]{""});
		dataGridMeta.setGroup(new String[]{""});
		dataGridMeta.setFieldLength(new int[]{-1});
		dataGridMeta.setFieldPrecision(new int[]{-1});
		dataGridMeta.setEmptyString(new boolean[]{false});
		dataGridMeta.setCurrency(new String[]{""});
		return dataGridMeta;
	}
	
	/**
	 * 获取记录关联组件，并且根据kettle画出的组件设置一些初始信息
	 * @return
	 */
	public static JoinRowsMeta getJoinRowsMeta(){
		JoinRowsMeta joinRowsMeta = new JoinRowsMeta();
		//设置临时目录
		joinRowsMeta.setDirectory("%%java.io.tmpdir%%");
		joinRowsMeta.setCacheSize(500);
		joinRowsMeta.setPrefix("out");
		Condition condition = new Condition();
		condition.setNegated(false);
		joinRowsMeta.setCondition(condition);
		return joinRowsMeta;
	}
	public static void initEnv() throws KettleException {
		EnvUtil.environmentInit();
		KettleEnvironment.init();
	}

	public static MongoDbInputData getMongoDBInputData() {
		MongoDbInputData stepDataInterface = new MongoDbInputData();

		return stepDataInterface;

	}
	/**
	 * 将生成的转换写入xml
	 * @param transMeta
	 * @param fileName
	 * @throws UnsupportedEncodingException
	 * @throws IOException
	 * @throws KettleException
	 */
	public static void writeToXML(TransMeta transMeta, String fileName)
			throws UnsupportedEncodingException, IOException, KettleException {
		String xml = transMeta.getXML();
		DataOutputStream dos = new DataOutputStream(new FileOutputStream(new File(fileName)));
		dos.write(xml.getBytes("UTF-8"));
		dos.close();
	}

	

	

	/**
	 * 该方法主要是因为第一层key映射过去到mysql中有些为关键字
	 * 
	 * 凡是出现如下的key，都设置为默认添加了下划线，即mongo中的name，映射到mysql中的_name
	 * 
	 */
	public static String mapColumnsToNonKey(String columnName) {
		Set<String> keyColumns = new HashSet<>();
		keyColumns.add("name");
		keyColumns.add("data");
		keyColumns.add("type");
		keyColumns.add("status");
		keyColumns.add("online");
		keyColumns.add("password");
		keyColumns.add("hour");
		keyColumns.add("minute");
		keyColumns.add("day");
		keyColumns.add("user");
		keyColumns.add("key");
		keyColumns.add("owner");
		keyColumns.add("plugin");
		keyColumns.add("logs");
		keyColumns.add("order");
		keyColumns.add("unique");
		keyColumns.add("error");
		keyColumns.add("group");
		keyColumns.add("session");
		keyColumns.add("signal");
		if (keyColumns.contains(columnName)) {
			return "_" + columnName;
		}
		return columnName;
	}
	
	/**
	 * 对自动创建的表，从collection映射到mysql的表名
	 * 需要对events表进行特殊处理，此处设置将mongo中events表放入到mysql中的eventsIncre表
	 * @param collectionName
	 * @return
	 */
	public static String mapIncreCollectionToTableName(String collectionName){
		List<String> lis = getIncreTables();
		for(String incre:lis){
			if(collectionName.startsWith(incre)){
				if(incre.startsWith("events")){
					return "eventsIncre";
				}
				return incre;
			}
		}
		return "";
	}
	/**
	 * countly中自增表的前缀
	 * @return
	 */
	public static List<String> getIncreTablesFixed(){
		List<String> lis = null;
		if(lis==null){
			lis = new ArrayList<String>();
			lis.add("app_crashes");
			lis.add("app_crashgroups");
			lis.add("app_crashusers");
			lis.add("app_users");
			lis.add("app_viewdata");
			lis.add("app_views");
			lis.add("events");
			lis.add("logs");
			lis.add("metric_changes");
		}
		return lis;
	}

	/**
	 * countly自增表，可用来调节GeneIncreKtrTest中的输出
	 * @return
	 */
	public synchronized static List<String> getIncreTables(){
		List<String> lis = null;
		if(lis==null){
			lis = new ArrayList<String>();
			//lis.add("app_crashes");
			//lis.add("app_crashgroups");
			//lis.add("app_crashusers");
			//lis.add("app_users");
			//lis.add("app_viewdata");
			//由于app_views中的数据比较特殊，先暂时不使用
			//lis.add("app_views");
			//lis.add("events");
			lis.add("logs");//logs595f463dccd6f47ce18a5d53总会报错  Connection reset异常
			//lis.add("metric_changes");
		}
		return lis;
	}
	/**
	 * @throws Exception 
	 * 
	 */
	public static void executeTrans(TransMeta transMeta) throws Exception {
		Trans trans = new Trans(transMeta);
		// 不传入参数

		trans.execute(null);

		// 等待任务结束
		trans.waitUntilFinished();
		
		// 抛出异常  
        if (trans.getErrors() > 0) {  
            throw new Exception(  
                    "There are errors during transformation exception!(传输过程中发生异常)");  
        } 

	}
	
	

}
